/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package wikiedits;

import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditEvent;
import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditsSource;

public class WikipediaAnalysisToKafka {

	/**
	 * $ mvn clean package
	 * $ mvn exec:java -Dexec.mainClass=wikiedits.WikipediaAnalysis
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<WikipediaEditEvent> edits = env.addSource(new WikipediaEditsSource());

		KeyedStream<WikipediaEditEvent, String> keyedEdits = edits.keyBy(new KeySelector<WikipediaEditEvent, String>() {
			@Override
			public String getKey(WikipediaEditEvent wikipediaEditEvent) throws Exception {
				return wikipediaEditEvent.getUser();
			}
		});

		DataStream<Tuple2<String, Long>> result = keyedEdits
				.timeWindow(Time.seconds(5))
				.fold(new Tuple2<>("", 0L), new FoldFunction<WikipediaEditEvent, Tuple2<String, Long>>() {
					@Override
					public Tuple2<String, Long> fold(Tuple2<String, Long> acc, WikipediaEditEvent event) {
						acc.f0 = event.getUser();
						acc.f1 += event.getByteDiff();
						return acc;
					}
				});

		result.map(new MapFunction<Tuple2<String,Long>, String>() {
					@Override
					public String map(Tuple2<String, Long> tuple) {
						return tuple.toString();
					}
				})
				.addSink(new FlinkKafkaProducer010<String>("106.14.183.23:9092", "test-wiki-result", new SimpleStringSchema()));



		env.execute("Flink Streaming WikipediaAnalysis");
	}
}
